package com.yjhh.hadoop.hdf;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;

/**
 * Created by hlantian on 2017/11/2.
 * Hadoop HDFS Java API操作
 */
public class HDFSApp {

    public static final String HDFS_PATH = "hdfs://Master:8020/";

    FileSystem fileSystem = null;

    Configuration configuration = null;

    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/hdfsapi2"));
    }

    @Test
    public void create() throws Exception {
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path(HDFS_PATH + "/a.txt"));
        fsDataOutputStream.write("hello hadoop".getBytes());
        fsDataOutputStream.flush();
        fsDataOutputStream.close();
    }

    @Test
    public void catFile() throws Exception {
        FSDataInputStream dataInputStream = fileSystem.open(new Path(HDFS_PATH + "/a.txt"));
        IOUtils.copyBytes(dataInputStream, System.out, 1024);
    }

    @Test
    public void rename() throws Exception {
        fileSystem.rename(new Path(HDFS_PATH + "/a.txt"), new Path(HDFS_PATH + "/aaa.txt"));
    }

    @Test
    public void copyFileToHdfs() throws Exception {
        fileSystem.copyFromLocalFile(new Path("D:\\report_data2.txt"), new Path("/hdfsapi2"));
    }

    @Test
    public void copyFileToLocal() throws Exception {
        fileSystem.copyToLocalFile(new Path(HDFS_PATH + "/hello.txt"), new Path("D:\\hello.txt"));
    }

    @Test
    public void listFiles() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/hdfsapi2"));
        for (FileStatus file : fileStatuses) {
            String isDir = file.isDirectory() ? "文件夹" : "文件";
            short replication = file.getReplication();   //获取副本个数
            long len = file.getLen();
            String filePath = file.getPath().toString();
            System.out.println(isDir+"   \t"+replication+"     \t"+len+"     \t"+filePath);
        }
    }

    @Test
    public void delete() throws Exception {
        fileSystem.delete(new Path("/hdfsapi3"),true);
    }

    @Test
    public void copyFileToHdfsProgress() throws Exception {
        fileSystem.copyFromLocalFile(new Path("F:\\谷歌下载目录\\jdk-8u131-linux-x64.tar.gz"), new Path("/hdfsapi2"));
        InputStream in = new BufferedInputStream(new FileInputStream(new File("F:\\谷歌下载目录\\jdk-8u131-linux-x64.tar.gz")));
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/hdfsapi2/jdk-8u131-linux-x64.tar.gz"), new Progressable() {

            public void progress() {
                System.out.print("*");     //进度
            }
        });

        IOUtils.copyBytes(in, fsDataOutputStream, 1024);
    }




    @Before
    public void setUp() throws Exception {
        System.out.println("HDFS.setUp");
        configuration = new Configuration();
        fileSystem = FileSystem.get(URI.create(HDFS_PATH), configuration);
    }

    @After
    public void tearDown() throws Exception {
        configuration = null;
        fileSystem = null;
        System.out.println("tearDown.tearDown");
    }
    
}
